package com.rocketmq.example;

import java.util.Date;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

public class E09_SendTransactionMessage {
    private static final String TOPIC = "TransactionMessage";


    public static void main(String[] args) {
        TransactionMQProducer producer = new TransactionMQProducer(ExampleMQConfig.TOPIC.getName());
        producer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());

        // 指定同步模式下,失败重试次数
        producer.setRetryTimesWhenSendFailed(5);
        // 设置事务监听器
        producer.setTransactionListener(new TransactionListener() {

            /**
             * COMMIT_MESSAGE：提交消息，这个消息由prepared状态进入到commited状态，消费者可以消费这个消息；
             * ROLLBACK_MESSAGE：回滚，这个消息将被删除，消费者不能消费这个消息；
             * NKNOW：未知，这个状态有点意思，如果返回这个状态，这个消息既不提交，也不回滚，还是保持prepared状态，而最终决定这个消息命运的，是checkLocalTransaction这个方法。默认在 executeLocalTransaction 提交 UNKNOW 一分钟后执行  checkLocalTransaction
             */

            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                System.err.println("提交事务消息" + new Date());
                return LocalTransactionState.UNKNOW;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.err.println("检查事务消息"+ new Date());
                return LocalTransactionState.COMMIT_MESSAGE;
            }

        });

        // 启动实例
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        for (int i = 0; i < 5; i++) {
            // 实例化消息对象
            Message message = new Message(TOPIC, "tagA", "msg-" + i, ("事务消息发送" + new Date()+ ":" + i).getBytes());
            // 发送消息
            SendResult sendResult = null;
            try {
                sendResult = producer.sendMessageInTransaction(message, i);
            } catch (MQClientException e) {
                e.printStackTrace();
            }
            System.out.printf("%s%n", sendResult);
        }

        // 关闭实例
        // producer.shutdown();
    }
}
